#include "mysql_task.h"

#include <filesystem>
#include <workflow/Workflow.h>
#include <workflow/WFTaskFactory.h>
#include <workflow/WFMySQLConnection.h>
#include <workflow/MySQLResult.h>
#include <workflow/WFFacilities.h>
#include <cstring>
#include <fstream>
#include <cstdlib>
#include <cppbase/utility.h> //get app global config
#include <brpc/http_status_code.h>
#include <butil/logging.h>
#define MYSQL_RETRY 1
using namespace protocol;
#define ASSERT_MYSQL_OK                                                                                            \
	DBResult *res = (DBResult *)(series_of(task)->get_context());                                                  \
	if (task->get_state() != WFT_STATE_SUCCESS)                                                                    \
	{                                                                                                              \
		res->error_code = brpc::HTTP_STATUS_INTERNAL_SERVER_ERROR;                                                 \
		res->error_str = WFGlobal::get_error_string(task->get_state(), task->get_error());                         \
		LOG(ERROR) << "mysql task failed,wf task state:" << task->get_state() << ",error text:" << res->error_str; \
		return;                                                                                                    \
	}                                                                                                              \
	else                                                                                                           \
	{                                                                                                              \
		res->error_code = 0;                                                                                       \
	}

namespace WFTask
{

	WFMySQLConnection *get_mysql_conn()
	{
		static WFMySQLConnection conn(1);
		static bool inited = false;
		if (!inited)
		{
			conn.init(AppGlobal::instance()->get_config("mysql_uri"));
			inited = true;
		}
		return &conn;
	}
	void commit_mysql(WFMySQLTask *task, DB_cb_t cb)
	{
		DBResult dbresult;
		auto *series = Workflow::create_series_work(task,
													[&dbresult, cb](const SeriesWork *series)
													{
														cb(dbresult.error_code, dbresult.error_str);
													});
		series->set_context(&dbresult);
		series->start();
	}
	void handler_CollectionList(WFMySQLTask *task)
	{
		ASSERT_MYSQL_OK;
		frpcType::CollectionArray *contracta = (frpcType::CollectionArray *)task->user_data;
		MySQLResultCursor cursor(task->get_resp());
		std::vector<MySQLCell> arr;

		while (cursor.fetch_row(arr))
		{
			auto c = contracta->add_collection();
			c->set_id(arr[0].as_int());
			c->set_name(arr[1].as_string());
			c->set_type_id(arr[2].as_int());
			c->set_count(arr[3].as_int());
			c->set_about(arr[4].as_string());
		}
	}

	void GetCollectionList(uint64_t *userid, frpcType::CollectionArray *response, DB_cb_t cb)
	{
		//auto task = WFTaskFactory::create_mysql_task(AppGlobal::instance()->get_config("mysql_uri"), MYSQL_RETRY, handler_CollectionList);
		std::string q = "select id,name,type_id,count,about from collection_info where id<10 or owner_id=" + std::to_string(*userid);
		auto task = get_mysql_conn()->create_query_task(q, handler_CollectionList);
		task->user_data = response;
		commit_mysql(task, cb);
	}
	void handler_ExchangeList(WFMySQLTask *task)
	{
		ASSERT_MYSQL_OK;
		frpcType::ExchangeArray *excharray = (frpcType::ExchangeArray *)task->user_data;
		MySQLResultCursor cursor(task->get_resp());
		std::vector<MySQLCell> arr;

		while (cursor.fetch_row(arr))
		{
			auto c = excharray->add_exchange();
			c->set_id(arr[0].as_int());
			c->set_code(arr[1].as_string());
			c->set_name(arr[2].as_string());
			c->set_currency(arr[3].as_string());
		}
	}
	void GetExchangeList(frpcType::ExchangeArray *response, DB_cb_t cb)
	{
		std::string q = "select id,code,name,currency from exchange_info";
		auto task = get_mysql_conn()->create_query_task(q, handler_ExchangeList);
		task->user_data = response;
		commit_mysql(task, cb);
	}

	void handler_CollectionContent(WFMySQLTask *task)
	{
		ASSERT_MYSQL_OK;
		MySQLResultCursor cursor(task->get_resp());
		std::vector<MySQLCell> arr;
		frpcType::ContractArray *ca = (frpcType::ContractArray *)task->user_data;
		if (cursor.get_field_count() > 5) //is system collection response
		{

			while (cursor.fetch_row(arr))
			{
				auto c = ca->add_contract();
				c->set_id(arr[0].as_int());
				c->set_name(arr[1].as_string());
				c->set_symbol(arr[2].as_string());
				c->set_exchange_id(arr[3].as_int());
				c->set_type_id(arr[4].as_int());
				c->set_company_id(arr[5].as_int());
			}
		}
		else
		{ //query user defined collection
			while (cursor.fetch_row(arr))
			{
				auto c = ca->add_contract();
				c->set_id(arr[0].as_int());
				c->set_exchange_id(arr[1].as_int());
				c->set_sort(arr[2].as_int());
			}
		}
	}
	void GetCollectionContent(uint64_t *userid, const frpcType::DefaultArg *req, frpcType::ContractArray *response, DB_cb_t cb)
	{
		char q[200];
		int coid = req->arg_int32(0);
		int start_index = req->arg_int32(1);
		int count = req->arg_int32(2);

		if (coid < 10)
		{
			int tid;
			tid=coid;
			sprintf(q, "SELECT id,name,symbol,exchange_id,type_id,company_id FROM exchange_sh where type_id=%d union SELECT id,name,symbol,exchange_id,type_id,company_id FROM exchange_sz where type_id=1 limit %d,%d",
					tid, start_index, count);
		}
		else
			sprintf(q, "SELECT contract_id,exchange_id,sort FROM collection_content where collection_id=%d limit %d,%d",
					coid, start_index, count);

		auto task = get_mysql_conn()->create_query_task(q, handler_CollectionContent);
		task->user_data = response;
		commit_mysql(task, cb);
	}
}